package com.bigdata

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1_WordCount2 {

  def main(args: Array[String]): Unit = {

    // 创建spark配置对象
    val conf: SparkConf = new SparkConf()

      /**
       * 指定指定模式为local本地模式
       * 1 指的是设置分区数为1，通用的设置
       */
      .setMaster("local[1]")
      .setAppName("WordCount") // 设置应用名称

    // 需要一个SparkContext对象
    val sc = new SparkContext(conf)
    // 设置日志的打印级别
    sc.setLogLevel("error")

    // 读取数据，分区数是2
    //    val linesRDD: RDD[String] = sc.textFile("data/words.txt", 2)
    val linesRDD: RDD[String] = sc.parallelize(Array("hello,world", "hello,spark", "hello,flink", "nihao,shijie"))
    //    val linesRDD: RDD[String] = sc.makeRDD(1 until 100)


    // 获取分区数, 就相当于并行度，driver里执行的
    println("getNumPartitions----" + linesRDD.getNumPartitions)

    // Driver -> Executor
    // Driver 就是复杂一个任务的包工头，一个任务对应一个Driver
    // Executor 是负责某个任务的农民工，一个Drvier可以对应多个Executor
    val num = 9527

    /**
     * spark 中提供了很多算子，比scala的集合函数还要丰富
     * rdd算子的代码是在哪里执行的？？？？
     * 在Executor中的线程池里执行的
     *
     */
    val wordRDD: RDD[String] = linesRDD.flatMap(line => {
      println("flatmap...." + num)
      line.split(",")
    })

    // 对每一个单词映射为一个("hadoop", 1), ("spark", 1)....
    // map算子可以调整分区吗？ 不能
    val pairRDD: RDD[(String, Int)] = wordRDD.map(word => (word, 1))

    //    pairRDD.foreach(println)


    // 按照key分区聚合，指定value的聚合逻辑
    // reduceByKey 是一个会产生shuffle的算子，会走磁盘
    // reduceByKey 会把相同的key进行聚合，传递一个函数，用来表示处理多个value的逻辑
    //    val result: RDD[(String, Int)] = pairRDD.reduceByKey((v1: Int, v2: Int) => {
    //      v1 + v2
    //    })
    // 对代码进行简化
    val result: RDD[(String, Int)] = pairRDD.reduceByKey(_ + _)

    // action算子触发任务的执行
    result.foreach(println)

  }

}
